package com.seventeen.monster.mq.proxy.server.proxy.ext;

import com.seventeen.monster.common.utils.StringUtils;
import com.seventeen.monster.mq.proxy.server.proxy.consumer.ConsumeStatus;
import com.seventeen.monster.mq.proxy.server.proxy.consumer.ConsumerTypes;
import com.seventeen.monster.mq.proxy.server.proxy.MqClient;
import org.apache.rocketmq.acl.common.AclClientRPCHook;
import org.apache.rocketmq.acl.common.SessionCredentials;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.*;
import org.apache.rocketmq.client.consumer.rebalance.AllocateMessageQueueAveragely;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.apache.rocketmq.remoting.RPCHook;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;
import java.util.Objects;

/**
 * 4.x 消费者
 *
 * @author seventeen
 */
public class DefaultExtMqPushConsumer extends BaseExtMqConsumer {
    private static final Logger LOGGER = LoggerFactory.getLogger(DefaultExtMqPushConsumer.class);
    private DefaultMQPushConsumer consumer;

    private RPCHook getAclRpcHook(String accessKey, String secretKey) {
        return new AclClientRPCHook(new SessionCredentials(accessKey, secretKey));
    }

    /**
     * 初始化
     */
    public void init() {
        try {
            String accessKey = getAccessKey();
            String secretKey = getSecretKey();
            if (StringUtils.isNotBlank(accessKey) && StringUtils.isNotBlank(secretKey)) {
                consumer = new DefaultMQPushConsumer(getConsumerProxyGroup(), getAclRpcHook(accessKey, secretKey), new AllocateMessageQueueAveragely());
            } else {
                consumer = new DefaultMQPushConsumer(getConsumerProxyGroup());
            }
            consumer.setNamesrvAddr(getEndpoints());
            consumer.subscribe(getTopic(), getTags());
            consumer.setConsumeMessageBatchMaxSize(batchMaxSize);
            consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET);
            consumer.setMessageModel(MessageModel.CLUSTERING);
            this.register();
            consumer.start();
            LOGGER.info("{} started!", this);
        } catch (Exception e) {
            LOGGER.error("NewExtMQProducer start fail:{},{},", consumerProxyGroup, getEndpoints(), e);
        }
    }

    public void destroy() {
        LOGGER.info("{} shutdown!", this);
        if (consumer != null) {
            consumer.shutdown();
        }
    }


    private void register() {
        ConsumerTypes consumerTypes = ConsumerTypes.getByType(this.type);
        if (Objects.isNull(consumerTypes)) {
            LOGGER.error(" {} register error, type[{}] is wrong!", this, type);
        }
        switch (Objects.requireNonNull(consumerTypes)) {
            case CONCURRENTLY:
                this.concurrentlyRegister();
                break;
            case ORDERLY:
                this.orderlyRegister();
                break;
            default:
                break;
        }

    }

    private void concurrentlyRegister() {
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> list, ConsumeConcurrentlyContext consumeConcurrentlyContext) {
                for (MessageExt messageExt : list) {
                    String convertJson = ExtProxyMessage.convert(messageExt);
                    ConsumeStatus consumeStatus = MqClient.consumer(appUrl, convertJson);
                    if (consumeStatus == ConsumeStatus.FAIL) {
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
    }

    private void orderlyRegister() {
        consumer.registerMessageListener((MessageListenerOrderly) (list, consumeOrderlyContext) ->
        {
            for (MessageExt messageExt : list) {
                String convertJson = ExtProxyMessage.convert(messageExt);
                ConsumeStatus consumer = MqClient.consumer(appUrl, convertJson);
                if (consumer == ConsumeStatus.FAIL) {
                    return ConsumeOrderlyStatus.SUSPEND_CURRENT_QUEUE_A_MOMENT;
                }
            }

            return ConsumeOrderlyStatus.SUCCESS;
        });
    }

    @Override
    public String toString() {
        return "DefaultExtMqPushConsumer{" +
                "consumer=" + consumer +
                ", nameServerAddr='" + endpoints + '\'' +
                ", consumerProxyGroup='" + consumerProxyGroup + '\'' +
                ", batchMaxSize=" + batchMaxSize +
                ", topic='" + topic + '\'' +
                ", tags='" + tags + '\'' +
                ", appUrl='" + appUrl + '\'' +
                ", type='" + type + '\'' +
                ", version='" + version + '\'' +
                '}';
    }

}
